package com.example.producer;

import java.util.ArrayList;
import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;

public class KafakaConsumer {

	protected final Logger logger = LoggerFactory.getLogger(this.getClass());

	private static final String TPOIC = "test";

//	监听主题test，有消息就接收
	@KafkaListener(id = "id", topicPartitions = { @TopicPartition(topic = TPOIC, partitions = { "0" }) })
	public void listenPartition(ArrayList<ConsumerRecord<?, ?>> records) {
		logger.info("Id Listener, Thread ID: " + Thread.currentThread().getId());
		logger.info("Id records size " + records.size());
		for (ConsumerRecord<?, ?> record : records) {
			Optional<?> kafkaMessage = Optional.ofNullable(record.value());
			logger.info("Received: " + record);
			// 判断是否有值
			if (kafkaMessage.isPresent()) {
				Object message = record.value();
				String topic = record.topic();
				logger.info("p Received message={}", message);
			}
		}
	}
}
